package com.cdream.basic.mq.rocket;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson.JSON;
import com.cdream.basic.classiccase.miaosha.entity.StockLog;
import com.cdream.basic.classiccase.miaosha.service.OrderInfoService;
import com.cdream.basic.classiccase.miaosha.service.StockLogService;
import com.google.common.collect.ImmutableMap;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 异步扣减数据库库存处理3大问题
 * 1. 异步消息发送失败
 * 2. 扣减操作执行失败
 * 3. 下单失败无法正确回补库存
 *
 * @author cdream
 * @date 2021/7/24
 */
@Component
public class RocketMqTransProducer {

    private TransactionMQProducer transactionMQProducer;
    @Value("${mq.nameserver.addr}")
    private String nameAddr;
    @Value("${mq.topicname}")
    private String topicName;
    @Autowired
    private OrderInfoService orderInfoService;
    @Autowired
    private StockLogService stockLogService;

    @PostConstruct
    public void init() throws MQClientException {
        // 做mq producer的初始化
        // producer的name没有什么意义，只是作为一个标识
        // comsumer的标识是有意义的，会将comsumer分成一个组
        transactionMQProducer = new TransactionMQProducer("transaction_producer_group");
        transactionMQProducer.setNamesrvAddr(nameAddr);
        transactionMQProducer.start();

        transactionMQProducer.setTransactionListener(new TransactionListener() {
            // 这里会向消息中间件发送消息，告诉执行成功或失败
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object arg) {
                Integer promoId = (Integer) ((Map) arg).get("promoId");
                Integer amount = (Integer) ((Map) arg).get("amount");
                Integer itemId = (Integer) ((Map) arg).get("itemId");
                Integer userId = (Integer) ((Map) arg).get("userId");
                String stockLogId = (String) ((Map) arg).get("stockLogId");
                try {
                    orderInfoService.createOrderByTransMq(userId, itemId, promoId, amount, stockLogId);
                } catch (RuntimeException e) {
                    // 无流水日志的情况下存在的问题，订单提交成功，确认ack未发送出去，直接断链了，会导致一致性问题
                    e.printStackTrace();
                    // 设置stocklog为回滚状态
                    StockLog stockLog = stockLogService.getById(stockLogId);
                    if (stockLog == null) {
                        throw new RuntimeException("库存数据异常");
                    }
                    stockLog.setStatus(3);
                    stockLogService.updateById(stockLog);
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                String jsonString = new String(messageExt.getBody());
                Map<String, Object> map = JSON.parseObject(jsonString, Map.class);
                Integer itemId = (Integer) map.get("itemId");
                Integer amount = (Integer) map.get("amount");
                String stockLogId = (String) map.get("stockLogId");
                StockLog stockLog = stockLogService.getById(stockLogId);
                if (stockLog == null) {
                    return LocalTransactionState.UNKNOW;
                }
                if (stockLog.getStatus() == 2) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                if (stockLog.getStatus() == 1) {
                    return LocalTransactionState.UNKNOW;
                }
                // 如果队列中消息一只是UNKNOWN状态,则会调用这个方法进行询问
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        });
    }

    public Boolean transAsyncReduceStock(Integer userId, Integer promoId, Integer itemId, Integer amount, String stockLogId) {
        // 投递给消费者
        Map<String, Object> bodyMap = ImmutableMap.of(
                "amount", amount,
                "itemId", itemId,
                "stockLogId", stockLogId);
        // 投递给LocalListener
        Map<String, Object> argsMap = ImmutableMap.of(
                "userId", userId,
                "amount", amount,
                "promoId", promoId,
                "itemId", itemId);
        Message message = new Message(topicName, "increase", JSON.toJSON(bodyMap).toString().getBytes(StandardCharsets.UTF_8));
        TransactionSendResult transactionSendResult;
        try {
            // 与非事务消息相比
            // 非事务消息无论如何都要发送
            // 事务形消息默任状态是prepare状态，向消息队列投送一个prepare的消息，本地执行localTransaction方法，
            // localTransaction是真正要执行的方法
            transactionSendResult = transactionMQProducer.sendMessageInTransaction(message, argsMap);
        } catch (MQClientException e) {
            return false;
        }
        return transactionSendResult.getLocalTransactionState() == LocalTransactionState.COMMIT_MESSAGE;
    }

    public static void findMinDifference(List<String> timePoints) {
        List<Integer> sort = timePoints.stream().map(a-> {
            String[] arr = a.split(":");
            int hour = Integer.parseInt(arr[0]) * 60;
            return hour + Integer.parseInt(arr[1]);
        }).sorted().collect(Collectors.toList());
        System.out.println("sort = " + sort);
    }

    public static void main(String[] args) {
        findMinDifference(CollUtil.newArrayList("23:59","00:00"));
    }
}
